package com.always.electrical.energy.mqtt;

import com.always.electrical.energy.constants.CMD;
import com.always.electrical.energy.mqtt.req.*;
import com.always.electrical.energy.mqtt.res.*;
import com.always.electrical.energy.utils.JsonApi;
import com.always.electrical.energy.utils.Util;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static org.eclipse.paho.client.mqttv3.MqttConnectOptions.MQTT_VERSION_3_1_1;

@Component
@Slf4j
public class MqttClients {

    private static final String PCS_DC_TOPIC = "TCMQTTIoT/terminal/cmdReq/Topscomm/Topscomm/1234/1234/";

    private final JsonApi jsonApi;

    private MqttClient mqttClient;

    private final Map<Integer, MqttTask> allTasks;

    private volatile int nextTaskId = 0;

    public MqttClients(@Value("${mqtt.url}") String url, @Value("${mqtt.clientId}") String clientId, @Value("${mqtt.timeOut}") Integer timeOut, JsonApi jsonApi) {
        this.allTasks = new ConcurrentHashMap<>();
        this.jsonApi = jsonApi;

        try {
            mqttClient = new MqttClient(url, clientId);
            mqttClient.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectionLost(Throwable throwable) {
                    log.error("connectionLost {}", throwable.getMessage());
                }

                // 接收数据
                @Override
                public void messageArrived(String topic, MqttMessage mqttMessage) {
                    receiveMqttMessage(topic, mqttMessage);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    log.info("send msg succeed topic is : {}", iMqttDeliveryToken.getTopics()[0]);
                }

                @Override
                public void connectComplete(boolean reconnect, String serverUri) {
                    try {
                        log.info("connectComplete reconnect {} {}", reconnect, serverUri);
                        mqttClient.subscribe("+/terminal/cmdRsp/Topscomm/#");
                    } catch (Exception e) {
                        log.error("connectComplete reconnect:{}", e.getMessage());
                    }
                }
            });

            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setUserName(null);
            mqttConnectOptions.setPassword(new char[]{});
            mqttConnectOptions.setMqttVersion(MQTT_VERSION_3_1_1);
            mqttConnectOptions.setCleanSession(false);
            mqttConnectOptions.setAutomaticReconnect(true);
            mqttConnectOptions.setConnectionTimeout(10);
            mqttConnectOptions.setKeepAliveInterval(30);
            mqttConnectOptions.setMaxInflight(100);
            mqttClient.connect(mqttConnectOptions);

            new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder().setNameFormat("mqtt-clear-scheduled-%d").build())
                    .scheduleWithFixedDelay(() -> {
                        for (MqttTask<?> mqttTask : allTasks.values()) {
                            if (System.currentTimeMillis() - mqttTask.getCreateTime() > timeOut * 1000L) {
                                allTasks.remove(mqttTask.mqttId());
                                log.info("timeOutTask {}{}allTasksSize {}", mqttTask.mqttId(), System.lineSeparator(), allTasks.size());
                                if (mqttTask.getCallBack() != null) {
                                    mqttTask.getCallBack().fail();
                                }
                            }
                        }
                    }, timeOut, 1, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("connect", e);
        }
    }

    // 接收数据
    private void receiveMqttMessage(String topic, MqttMessage mqttMessage) {
        try {
            String content = new String(mqttMessage.getPayload(), StandardCharsets.UTF_8);
            log.info("receive topic {}{}message:{}", topic, System.lineSeparator(), content);

            String cmd;
            String[] split = topic.split("/");
            if (Objects.equals(split[split.length - 1], "commandResponse")) {
                cmd = Util.getCmd(content);
            } else {
                JsonNode body = jsonApi.readTree(content).path("body");
                cmd = body.path("serviceProperties").fieldNames().next();
                content = jsonApi.toJson(body);
            }
            log.info("cmd={}", cmd);
            switch (cmd) {
                case CMD.CONTROL_RUNNING_STATE_GET:
                    executeTask(jsonApi.fromJsonTypeRef(content, ControlRunningStateGetRes.class));
                    break;
                case CMD.CONTROL_MACHINE_MODE_GET:
                    executeTask(jsonApi.fromJsonTypeRef(content, ControlMachineModeGetRes.class));
                    break;
                case CMD.CONTROL_RUNNING_STATE_SET:
                    executeTask(jsonApi.fromJsonTypeRef(content, ControlRunningStateSetRes.class));
                    break;
                case CMD.THRESHOLD_POSITIVE:
                    executeTask(jsonApi.fromJsonTypeRef(content, ThresholdPositiveRes.class));
                    break;
                case CMD.THRESHOLD_NEGATIVE:
                    executeTask(jsonApi.fromJsonTypeRef(content, ThresholdNegativeRes.class));
                    break;
                case CMD.THRESHOLD_CHARGE:
                    executeTask(jsonApi.fromJsonTypeRef(content, ThresholdChargeRes.class));
                    break;
                case CMD.THRESHOLD_DISCHARGE:
                    executeTask(jsonApi.fromJsonTypeRef(content, ThresholdDischargeRes.class));
                    break;
                case CMD.MICROGRID_GRIDST_OP_SET:
                    executeTask(jsonApi.fromJsonTypeRef(content, MicrogridGridStOpSetRes.class));
                    break;
                case CMD.PCS_STORAGE_RUN_STATE_GET:
                    executeTask(jsonApi.fromJsonTypeRef(content, PcsStorageRunningStateGetRes.class));
                    break;
                case CMD.PCS_RUN_STATE_MODE:
                    executeTask(jsonApi.fromJsonTypeRef(content, PcsRunningStateModeRes.class));
                    break;
                case CMD.PCS_STORAGE_MACHINE_STATE:
                    executeTask(jsonApi.fromJsonTypeRef(content, PcsStorageMachineStateRes.class));
                    break;
                case CMD.PCS_DC_GET:
                    executeTask(jsonApi.fromJsonTypeRef(content, PcsGetDcGetRes.class));
                    break;
                case CMD.STORAGE_DC_ON:
                    executeTask(jsonApi.fromJsonTypeRef(content, StorageDcOnRes.class));
                    break;
                case CMD.STORAGE_DC_OFF:
                    executeTask(jsonApi.fromJsonTypeRef(content, StorageDcOffRes.class));
                    break;
                case CMD.STORAGE_CLR_FLT:
                    executeTask(jsonApi.fromJsonTypeRef(content, StorageClrFltRes.class));
                    break;
                case CMD.STORAGE_W_SET:
                    executeTask(jsonApi.fromJsonTypeRef(content, StorageWSetRes.class));
                    break;
            }
        } catch (Exception e) {
            log.error("receiveMqttMessage", e);
        }
    }

    // 发送数据
    private <RES extends MqttResponse> MqttTask<RES> sendMqttMessage(BigDecimal tmlId, MqttTask<RES> mqttTaskReq) {
        try {
            int nextTaskId = getTaskId();
            while (allTasks.containsKey(nextTaskId)) {
                nextTaskId = getTaskId();
            }
            mqttTaskReq.setMid(nextTaskId);
            allTasks.put(mqttTaskReq.getMid(), mqttTaskReq);

            String topic = "/v1/devices/" + tmlId + "/command";
            String message = jsonApi.toJson(mqttTaskReq);
            log.info("send topic:{}{}message:{}", topic, System.lineSeparator(), message);
            mqttClient.publish(topic, new MqttMessage(message.getBytes(StandardCharsets.UTF_8)));
            return mqttTaskReq;
        } catch (Exception e) {
            log.error("sendMqttMessage", e);
        }
        return mqttTaskReq;
    }

    // 发送数据
    private <RES extends MqttResponse> MqttTask<RES> sendMqttMessageNew(String topic, MqttTask<RES> mqttTaskReq) {
        try {
            int nextTaskId = getTaskId();
            while (allTasks.containsKey(nextTaskId)) {
                nextTaskId = getTaskId();
            }
            mqttTaskReq.setRequestId(Integer.toString(nextTaskId));
            allTasks.put(nextTaskId, mqttTaskReq);

            String message = jsonApi.toJson(new MqttTaskWrapper<>(Integer.toString(nextTaskId), mqttTaskReq));
            log.info("send topic:{}{}message:{}", topic, System.lineSeparator(), message);
            mqttClient.publish(topic, new MqttMessage(message.getBytes(StandardCharsets.UTF_8)));
        } catch (Exception e) {
            log.error("sendMqttMessage", e);
        }
        return mqttTaskReq;
    }

    private <RES extends MqttResponse> void executeTask(MqttTaskRes<RES> mqttTaskRes) {
        log.info("executeTask {}", mqttTaskRes);
        Integer mqttId = mqttTaskRes.mqttId();
        MqttTask<RES> task = allTasks.get(mqttId);
        if (task != null) {
            allTasks.remove(mqttId);
            if (task.getCallBack() != null) {
                if (mqttTaskRes.getBody() != null) {
                    task.getCallBack().callBack(true, mqttTaskRes.getBody());
                } else if (mqttTaskRes.getServiceProperties() != null) {
                    task.getCallBack().callBack(true, mqttTaskRes.getServiceProperties());
                } else {
                    log.error("{} 返回结果为空", mqttId);
                    task.getCallBack().callBack(false, null);
                }
            }
        }
    }

    private synchronized int getTaskId() {
        nextTaskId++;
        if (nextTaskId > Integer.MAX_VALUE - 10) {
            nextTaskId = 1;
        }
        return nextTaskId;
    }

    // 设备查询 状态 召测
    public MqttTask<ControlRunningStateGetRes> sendControlRunningStateGetReq(BigDecimal tmlId, BigDecimal deviceId) {
        return sendMqttMessage(tmlId, new MqttTaskReq<ControlRunningStateGetReq, ControlRunningStateGetRes>()
                .setParas(new ControlRunningStateGetReq().setRunningState(""))
                .setCmd(CMD.CONTROL_RUNNING_STATE_GET).setDeviceId(String.valueOf(deviceId)));
    }

    // 设备查询 模式 召测
    public MqttTask<ControlMachineModeGetRes> sendControlMachineModeGetReq(BigDecimal tmlId, BigDecimal deviceId) {
        return sendMqttMessage(tmlId, new MqttTaskReq<ControlMachineModeGetReq, ControlMachineModeGetRes>()
                .setParas(new ControlMachineModeGetReq().setMachineMode(""))
                .setCmd(CMD.CONTROL_MACHINE_MODE_GET).setDeviceId(String.valueOf(deviceId)));
    }

    // 设备控制 状态 下发
    public MqttTask<ControlRunningStateSetRes> sendControlRunningStateSetReq(String nodeId, String state) {
        return sendMqttMessageNew("{app98}/terminal/cmdReq/Topscomm/smartchip/SVG/SVG/" + nodeId,
                new MqttTaskReq<ControlRunningStateSetReq, ControlRunningStateSetRes>()
                        .setCmdContent(new ControlRunningStateSetReq().setRunningState(state))
                        .setMethod(CMD.CONTROL_RUNNING_STATE_SET));
    }

    // 阈值调控 正向重过载阈值 召测/下发
    public MqttTask<ThresholdPositiveRes> sendThresholdPositiveReq(BigDecimal tmlId, BigDecimal deviceId, String positive) {
        return sendMqttMessage(tmlId, new MqttTaskReq<ThresholdPositiveReq, ThresholdPositiveRes>()
                .setParas(new ThresholdPositiveReq().setPositive(positive))
                .setCmd(CMD.THRESHOLD_POSITIVE).setDeviceId(String.valueOf(deviceId)));
    }

    // 阈值调控 反向重过载阈值 召测/下发
    public MqttTask<ThresholdNegativeRes> sendThresholdNegativeReq(BigDecimal tmlId, BigDecimal deviceId, String negative) {
        return sendMqttMessage(tmlId, new MqttTaskReq<ThresholdNegativeReq, ThresholdNegativeRes>()
                .setParas(new ThresholdNegativeReq().setNegative(negative))
                .setCmd(CMD.THRESHOLD_NEGATIVE).setDeviceId(String.valueOf(deviceId)));
    }

    // 阈值调控 储能充电阈值 召测/下发
    public MqttTask<ThresholdChargeRes> sendThresholdChargeReq(BigDecimal tmlId, BigDecimal deviceId, String charge) {
        return sendMqttMessage(tmlId, new MqttTaskReq<ThresholdChargeReq, ThresholdChargeRes>()
                .setParas(new ThresholdChargeReq().setCharge(charge))
                .setCmd(CMD.THRESHOLD_CHARGE).setDeviceId(String.valueOf(deviceId)));
    }

    // 阈值调控 储能放电阈值 召测/下发
    public MqttTask<ThresholdDischargeRes> sendThresholdDisChargeReq(BigDecimal tmlId, BigDecimal deviceId, String discharge) {
        return sendMqttMessage(tmlId, new MqttTaskReq<ThresholdDischargeReq, ThresholdDischargeRes>()
                .setParas(new ThresholdDischargeReq().setDischarge(discharge))
                .setCmd(CMD.THRESHOLD_DISCHARGE).setDeviceId(String.valueOf(deviceId)));
    }

    // 微网系统 计划并离网 下发
    public MqttTask<MicrogridGridStOpSetRes> sendMicrogridGridStOpSetReq(BigDecimal tmlId, BigDecimal deviceId, String op) {
        return sendMqttMessage(tmlId, new MqttTaskReq<MicrogridGridStOpSetReq, MicrogridGridStOpSetRes>()
                .setParas(new MicrogridGridStOpSetReq().setOp(op))
                .setCmd(CMD.MICROGRID_GRIDST_OP_SET).setDeviceId(String.valueOf(deviceId)));
    }


    // PCS 储能 运行状态 召测
    public MqttTask<PcsStorageRunningStateGetRes> sendPcsStorageRunningStateGetReq(String nodeId, String runState) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<PcsStorageRunningStateGetReq, PcsStorageRunningStateGetRes>()
                        .setCmdContent(new PcsStorageRunningStateGetReq().setRunningState(runState)));
    }

    // PCS 运行状态 运行模式 下发
    public MqttTask<PcsRunningStateModeRes> sendPcsRunningStateModeReq(String nodeId, String runStateMode) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<PcsRunningStateModeReq, PcsRunningStateModeRes>()
                        .setCmdContent(new PcsRunningStateModeReq().setRunningStateMode(runStateMode)));
    }

    // PCS 储能 整机状态 召测/下发
    public MqttTask<PcsStorageMachineStateRes> sendPcsStorageMachineStateReq(String nodeId, String machineState) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<PcsStorageMachineStateReq, PcsStorageMachineStateRes>()
                        .setCmdContent(new PcsStorageMachineStateReq().setMachineState(machineState)));
    }

    // PCS 直流侧电压 召测
    public MqttTask<PcsGetDcGetRes> sendPcsDcGetReq(String nodeId) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<PcsGetDcGetReq, PcsGetDcGetRes>()
                        .setCmdContent(new PcsGetDcGetReq().setDc("")));
    }

    // 储能 开机 下发
    public MqttTask<StorageDcOnRes> sendStorageDcOnReq(String nodeId) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<StorageDcOnReq, StorageDcOnRes>()
                        .setCmdContent(new StorageDcOnReq().setDcOn("1")));
    }

    // 储能 关机 下发
    public MqttTask<StorageDcOffRes> sendStorageDcOffReq(String nodeId) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<StorageDcOffReq, StorageDcOffRes>()
                        .setCmdContent(new StorageDcOffReq().setDcOff("1")));
    }

    // 储能 清除故障 下发
    public MqttTask<StorageClrFltRes> sendStorageClrFltReq(String nodeId) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<StorageClrFltReq, StorageClrFltRes>()
                        .setCmdContent(new StorageClrFltReq().setClrFlt("1")));
    }

    // 储能 设备侧功率 下发
    public MqttTask<StorageWSetRes> sendStorageWSetReq(String nodeId, String w) {
        return sendMqttMessageNew(PCS_DC_TOPIC + nodeId,
                new MqttTaskReq<StorageWSetReq, StorageWSetRes>()
                        .setCmdContent(new StorageWSetReq().setW(w)));
    }
}
